OperatorUtils

Utility helpers for operator-related queries.

Example

from human_protocol_sdk.constants import ChainId
from human_protocol_sdk.operator import OperatorUtils, OperatorFilter

print(
    OperatorUtils.get_operators(
        OperatorFilter(chain_id=ChainId.POLYGON_AMOY, roles=["Job Launcher"])
    )
)

OperatorData

OperatorData(
    chain_id,
    id,
    address,
    amount_jobs_processed,
    reputation_networks,
    staked_amount=None,
    locked_amount=None,
    locked_until_timestamp=None,
    withdrawn_amount=None,
    slashed_amount=None,
    role=None,
    fee=None,
    public_key=None,
    webhook_url=None,
    website=None,
    url=None,
    job_types=None,
    registration_needed=None,
    registration_instructions=None,
    name=None,
    category=None,
)

Represents operator information retrieved from the subgraph.

Attributes:

Name	Type	Description
chain_id	ChainId	Chain where the operator is registered.
id	str	Unique operator identifier.
address	str	Operator's Ethereum address.
staked_amount	Optional[int]	Amount staked by the operator.
locked_amount	Optional[int]	Amount currently locked.
locked_until_timestamp	Optional[int]	Time in milliseconds until locked amount is released.
withdrawn_amount	Optional[int]	Total amount withdrawn.
slashed_amount	Optional[int]	Total amount slashed.
amount_jobs_processed	int	Number of jobs launched/processed by the operator.
role	Optional[str]	Current role of the operator (e.g., "Job Launcher", "Recording Oracle").
fee	Optional[int]	Operator fee percentage.
public_key	Optional[str]	Operator's public key.
webhook_url	Optional[str]	Webhook URL for notifications.
website	Optional[str]	Operator's website URL.
url	Optional[str]	Operator URL.
job_types	List[str]	List of supported job types.
registration_needed	Optional[bool]	Whether registration is required.
registration_instructions	Optional[str]	Instructions for registration.
reputation_networks	List[str]	List of reputation network addresses.
name	Optional[str]	Operator name.
category	Optional[str]	Operator category.

Represents operator information returned from the subgraph.

Parameters:

Name	Type	Description	Default
chain_id	ChainId	Chain identifier.	required
id	str	Operator ID.	required
address	str	Operator address.	required
amount_jobs_processed	str	Jobs launched by the operator.	required
reputation_networks	Union[List[str], str]	List of reputation networks.	required
staked_amount	Optional[str]	Amount staked.	None
locked_amount	Optional[str]	Amount locked.	None
locked_until_timestamp	Optional[str]	Time (in seconds) until locked amount is released.	None
withdrawn_amount	Optional[str]	Amount withdrawn.	None
slashed_amount	Optional[str]	Amount slashed.	None
role	Optional[str]	Current role of the operator.	None
fee	Optional[str]	Operator fee.	None
public_key	Optional[str]	Public key.	None
webhook_url	Optional[str]	Webhook URL.	None
website	Optional[str]	Website URL.	None
url	Optional[str]	Operator URL.	None
job_types	Optional[Union[List[str], str]]	Supported job types.	None
registration_needed	Optional[bool]	Whether registration is needed.	None
registration_instructions	Optional[str]	Registration instructions.	None
name	Optional[str]	Operator name.	None
category	Optional[str]	Operator category.	None

OperatorFilter

OperatorFilter(
    chain_id,
    roles=[],
    min_staked_amount=None,
    order_by=None,
    order_direction=OrderDirection.DESC,
    first=10,
    skip=0,
)

Filter configuration for querying operators from the subgraph.

Attributes:

Name	Type	Description
chain_id	ChainId	Chain ID to request data from.
roles	List[str]	List of roles to filter by.
min_staked_amount	Optional[int]	Minimum staked amount to include operators.
order_by	Optional[str]	Property to order results by (e.g., "role", "stakedAmount").
order_direction	OrderDirection	Order direction (ascending or descending).
first	int	Number of items per page (1-1000).
skip	int	Number of items to skip for pagination.

Configure filtering options for operator queries.

Parameters:

Name	Type	Description	Default
chain_id	ChainId	Chain ID to request data.	required
roles	Optional[str]	Roles to filter by.	[]
min_staked_amount	int	Minimum amount staked to include.	None
order_by	Optional[str]	Property to order by, e.g., "role".	None
order_direction	OrderDirection	Order direction of results.	DESC
first	int	Number of items per page.	10
skip	int	Number of items to skip (for pagination).	0

Raises:

Type	Description
OperatorUtilsError	If chain ID or order direction is invalid.

OperatorUtils

Utility class providing operator-related query and data retrieval functions.

This class offers static methods to fetch operator data, including filtered operator lists, individual operator details, reputation network operators, and reward information from the Human Protocol subgraph.

get_operator staticmethod

get_operator(chain_id, operator_address, options=None)

Retrieve a single operator by their address.

Fetches detailed information about a specific operator from the subgraph.

Parameters:

Name	Type	Description	Default
chain_id	ChainId	Network where the operator is registered.	required
operator_address	str	Ethereum address of the operator.	required
options	Optional[SubgraphOptions]	Optional configuration for subgraph requests.	None

Returns:

Type	Description
Optional[OperatorData]	Operator data if found, otherwise None.

Raises:

Type	Description
OperatorUtilsError	If the chain ID is invalid or the operator address is malformed.

Example

from human_protocol_sdk.constants import ChainId
from human_protocol_sdk.operator import OperatorUtils

chain_id = ChainId.POLYGON_AMOY
operator_address = "0x62dD51230A30401C455c8398d06F85e4EaB6309f"

operator_data = OperatorUtils.get_operator(chain_id, operator_address)
if operator_data:
    print(f"Role: {operator_data.role}")
    print(f"Staked: {operator_data.staked_amount}")

get_operators staticmethod

get_operators(filter, options=None)

Retrieve a list of operators matching the provided filter criteria.

Queries the subgraph for operators that match the specified parameters including roles, minimum staked amount, and ordering preferences.

Parameters:

Name	Type	Description	Default
filter	OperatorFilter	Filter parameters including chain ID, roles, minimum staked amount, ordering, and pagination options.	required
options	Optional[SubgraphOptions]	Optional configuration for subgraph requests such as custom endpoints or timeout settings.	None

Returns:

Type	Description
List[OperatorData]	A list of operator records matching the filter criteria. Returns an empty list if no matches are found.

Example

from human_protocol_sdk.constants import ChainId
from human_protocol_sdk.operator import OperatorUtils, OperatorFilter

operators = OperatorUtils.get_operators(
    OperatorFilter(chain_id=ChainId.POLYGON_AMOY, roles=["Job Launcher"])
)
for operator in operators:
    print(f"{operator.address}: {operator.role}")

get_reputation_network_operators staticmethod

get_reputation_network_operators(
    chain_id, address, role=None, options=None
)

Retrieve operators registered under a specific reputation network.

Fetches all operators associated with a reputation oracle, optionally filtered by role.

Parameters:

Name	Type	Description	Default
chain_id	ChainId	Network where the reputation network exists.	required
address	str	Ethereum address of the reputation oracle.	required
role	Optional[str]	Optional role to filter operators (e.g., "Job Launcher").	None
options	Optional[SubgraphOptions]	Optional configuration for subgraph requests.	None

Returns:

Type	Description
List[OperatorData]	A list of operators registered under the reputation network. Returns an empty list if no operators are found.

Raises:

Type	Description
OperatorUtilsError	If the chain ID is invalid or the reputation address is malformed.

Example

from human_protocol_sdk.constants import ChainId
from human_protocol_sdk.operator import OperatorUtils

operators = OperatorUtils.get_reputation_network_operators(
    ChainId.POLYGON_AMOY,
    "0x62dD51230A30401C455c8398d06F85e4EaB6309f",
    role="Recording Oracle",
)
print(f"Found {len(operators)} operators")

get_rewards_info staticmethod

get_rewards_info(chain_id, slasher, options=None)

Retrieve rewards collected by a slasher address.

Fetches all reward events where the specified address acted as a slasher and received rewards for detecting misbehavior.

Parameters:

Name	Type	Description	Default
chain_id	ChainId	Network where the slasher operates.	required
slasher	str	Ethereum address of the slasher.	required
options	Optional[SubgraphOptions]	Optional configuration for subgraph requests.	None

Returns:

Type	Description
List[RewardData]	A list of rewards received by the slasher. Returns an empty list if no rewards are found.

Raises:

Type	Description
OperatorUtilsError	If the chain ID is invalid or the slasher address is malformed.

Example

from human_protocol_sdk.constants import ChainId
from human_protocol_sdk.operator import OperatorUtils

rewards_info = OperatorUtils.get_rewards_info(
    ChainId.POLYGON_AMOY,
    "0x62dD51230A30401C455c8398d06F85e4EaB6309f",
)
total_rewards = sum(reward.amount for reward in rewards_info)
print(f"Total rewards: {total_rewards}")

OperatorUtilsError

Bases: Exception

Exception raised when errors occur during operator operations.

RewardData

RewardData(escrow_address, amount)

Represents a reward distributed to a slasher.

Attributes:

Name	Type	Description
escrow_address	str	Address of the escrow that generated the reward.
amount	int	Reward amount in token's smallest unit.