RPC query engine

Shell-level endpoints

class pytezos.rpc.shell.BlockInjectionQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(block, _async=False, force=False, chain=None)[source]

Inject a block in the node and broadcast it. The operations embedded in blockHeader might be pre-validated using a contextual RPCs from the latest block (e.g. ‘/blocks/head/context/preapply’).

Parameters
  • block

    Json input {

    ”data”: <hex-encoded block header>, “operations”: [ [ { “branch”: <block_hash>, “data”: <hex-encoded operation> } … ] … ]

    }

  • _async – By default, the RPC will wait for the block to be validated before answering, set True if you don’t want to.

  • force

  • chain – Optionally you can specify the chain

Returns

ID of the block

class pytezos.rpc.shell.ChainQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
watermark()[source]

Chain watermark, hex encoded.

class pytezos.rpc.shell.ConnectionQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

delete(wait=False)[source]
property path
class pytezos.rpc.shell.DescribeQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(recurse=True)[source]

Get RPCs documentation and input/output schema.

Parameters

recurse – Show information for child elements, default is True.

In some cases doesn’t work without this flag.

property path
class pytezos.rpc.shell.InvalidBlockQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

delete()[source]
property path
class pytezos.rpc.shell.MempoolQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(configuration)[source]

Set operation filter rules.

Parameters

configuration – a JSON dictionary, known keys are minimal_fees, minimal_nanotez_per_gas_unit, minimal_nanotez_per_byte

class pytezos.rpc.shell.MonitorQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(*args, **kwargs)[source]

Call self as a function.

property path
class pytezos.rpc.shell.NetworkItems(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(_filter=None)[source]

Call self as a function.

property path
class pytezos.rpc.shell.NetworkLogQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(monitor=False)[source]

Call self as a function.

property path
class pytezos.rpc.shell.OperationInjectionQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(operation, _async=False, chain=None)[source]

Inject an operation in node and broadcast it. The signedOperationContents should be constructed using a contextual RPCs from the latest block and signed by the client.

Parameters
  • operation – Hex-encoded operation data or bytes

  • _async – By default, the RPC will wait for the operation to be (pre-)validated before answering, set True if you don’t want to.

  • chain – Optionally you can specify the chain

Returns

ID of the operation

class pytezos.rpc.shell.PendingOperationsQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
class pytezos.rpc.shell.ProtocolInjectionQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(protocol, _async=False, force=False)[source]

Inject a protocol in node.

Parameters
  • protocol

    Json input {

    ”expected_env_version”: <integer>, “components”: [{

    ”name”: <unistring>, “interface”?: <hex-encoded data>, “implementation”: <hex-encoded data> }

    ]}

    }

  • _async

  • force

Returns

ID of the protocol

class pytezos.rpc.shell.ResponseGenerator(res: requests.models.Response)[source]
class pytezos.rpc.shell.ShellQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property block

Cached head block, useful if you just want to explore things.

property blocks

Shortcut for chains.main.blocks

property contracts

Shortcut for head.context.contracts

property cycles

Operate on cycles rather than blocks.

property head

Shortcut for blocks.head

property mempool

Shortcut for chains.main.mempool

property path
property voting_periods

Operate on voting periods rather than blocks.

wait_next_block(block_hash=None)[source]

Wait until next block is finalized.

Parameters

block_hash – Current block hash (optional). If not set, current head is used.

pytezos.rpc.shell.make_operation_result(**kwargs)[source]

Protocol-specific endpoints

class pytezos.rpc.protocol.BigMapGetQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(query: dict)[source]

Access the value associated with a key in the big map storage of the michelson.

:param query
{

key: { $key_type : <key> }, type: { “prim” : $key_prim }

} $key_type: Provided key encoding, e.g. “string”, “bytes” for hex-encoded string, “int” key_prim: Expected high-level data type, e.g. “address”, “nat”, “mutez” (see storage section in code)

Returns

Micheline expression

class pytezos.rpc.protocol.BlockQuery(*args, **kwargs)[source]
__call__(**params)

Call self as a function.

property baker

Query block producer (baker).

Return type

ContractQuery

cycle() → int[source]

Get cycle for this block from metadata.

level() → int[source]

Get level for this block from metadata.

property path
property predecessor

Query previous block.

Return type

BlockQuery

voting_period() → int[source]

Get voting period for this block from metadata.

class pytezos.rpc.protocol.BlocksQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(length=1, head=None, min_date=None)[source]

Lists known heads of the blockchain sorted with decreasing fitness. Optional arguments allows to returns the list of predecessors for known heads or the list of predecessors for a given list of blocks.

Parameters
  • length – The requested number of predecessors to returns (per requested head).

  • head – An empty argument requests blocks from the current heads. A non empty list allow to request specific fragment of the chain.

  • min_date – When min_date is provided, heads with a timestamp before min_date are filtered out

Return type

list[list[str]]

property current_cycle

Get block range for the current cycle.

Return type

BlockSliceQuery

property current_voting_period

Get block range for the current voting period.

Return type

BlockSliceQuery

property path
class pytezos.rpc.protocol.ContextRawBytesQuery(*args, **kwargs)[source]
__call__(depth=1) → dict[source]

Return the raw context.

Parameters

depth – Context is a tree structure, default depth is 1

property path
class pytezos.rpc.protocol.ContextRawJsonQuery(*args, **kwargs)[source]
__call__(**params)

Call self as a function.

property path
class pytezos.rpc.protocol.ContextSeedQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post()[source]

Get seed of the cycle to which the block belongs.

class pytezos.rpc.protocol.ContractQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

code()[source]

Get contract code.

Returns

Micheline expression

count() → Iterator[source]

Get contract counter iterator: it returns incremented value on each call.

property path
public_key() → str[source]

Retrieve the contract manager’s public key (base58 encoded)

class pytezos.rpc.protocol.EndorsingPower(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(endorsement_operation)[source]

Get the endorsing power of an endorsement operation, that is, the number of slots that the op has.

:param endorsement_operation
{

“branch”: $block_hash, “contents”: [ $operation.alpha.contents … ], “signature”: $Signature

}

class pytezos.rpc.protocol.OperationListListQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property anonymous
seed_nonce_revelation, double_endorsement_evidence,

double_baking_evidence, activate_account.

Return type

OperationListQuery

Type

Operations with content of type

property endorsements

endorsement.

Return type

OperationListQuery

Type

Operations with content of type

find_ballots(proposal_id=None) → list[source]

Find operations of kind ballot.

Parameters

proposal_id – Proposal hash (optional)

Returns

list of operation contents

find_origination(contract_id)[source]

Find origination of the contract.

Parameters

contract_id – Contract ID (KT-address)

Returns

operation content

find_upvotes(proposal_id) → list[source]

Find operations of kind proposal for given proposal.

Parameters

proposal_id – Proposal hash (base58)

Returns

list of operation contents

property managers

reveal, transaction, origination, delegation.

Return type

OperationListQuery

Type

Operations with content of type

property path
property votes

proposal, ballot.

Return type

OperationListQuery

Type

Operations with content of type

class pytezos.rpc.protocol.OperationQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
unsigned() → dict[source]

Get operation group data without metadata and signature.

class pytezos.rpc.protocol.ProposalQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__() → int[source]

Roll count for this proposal.

property path
class pytezos.rpc.protocol.ProposalsQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
pytezos.rpc.protocol.to_timestamp(v)[source]

RPC Helpers

class pytezos.rpc.helpers.BakingRightsQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(level=None, cycle=None, delegate=None, max_priority=None, _all=None)[source]

Retrieves the list of delegates allowed to bake a block. By default, it gives the best baking priorities for bakers that have at least one opportunity below the 64th priority for the next block.

Parameters
  • level – Specify the (valid) level in the past or future at which the baking rights have to be returned.

  • cycle – Specify the (valid) levels in the past or future at which the baking rights have to be returned.

  • delegate – Restrict the results to the given delegates.

  • max_priority

  • _all – If parameter all is set, all the baking opportunities for each baker at each level are returned, instead of just the first one.

Returns

Returns the list of baking slots. Also returns the minimal timestamps that correspond to these slots. The timestamps are omitted for levels in the past, and are only estimates for levels later that the next block, based on the hypothesis that all predecessor blocks were baked at the first priority.

property path
class pytezos.rpc.helpers.ForgeBlockHeaderQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(block_header)[source]

Forge block header.

Parameters

block_header

Json input {

”level”: <integer>, “proto”: <integer>, “predecessor”: <block hash>, “timestamp”: <timestamp>, “validation_pass”: <integer>, “operations_hash”: <Operation_list_list_hash>, “fitness”: <integer (hex)>, “context”: <Context_hash>, “protocol_data”: <hex encoded>

}

Returns

{ “block”: <hex encoded> }

class pytezos.rpc.helpers.ForgeOperationsQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(operation)[source]

Get raw bytes encoding of an unsigned operation

Parameters

operation

Json input {

”branch”: <block hash>, “contents”: [{ <operation_content> }]

}

Returns

Hex string

class pytezos.rpc.helpers.ForgeProtocolDataQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(protocol_data)[source]

Forge the protocol-specific part of a block header.

Parameters

protocol_data

Json input {

”priority”: <integer>, “nonce_hash”: <Base58 nonce hash>, // optional “proof_of_work_nonce”: <hex encoded> // optional

}

Returns

{ “protocol_data”: <hex encoded> }

class pytezos.rpc.helpers.ParseBlockQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(block_header)[source]

Retrieves protocol-specific part of a block header and signature.

Parameters

block_header

Json input {

”level”: <integer>, “proto”: <integer>, “predecessor”: <block hash>, “timestamp”: <timestamp>, “validation_pass”: <integer>, “operations_hash”: <Operation_list_list_hash>, “fitness”: <integer (hex)>, “context”: <Context_hash>, “protocol_data”: <hex encoded>

}

Returns

Json object

class pytezos.rpc.helpers.ParseOperationsQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(operations)[source]

Extracts contents and signatures from the forged and optionally signed operations (bulk).

Parameters

operations

Json input {

”operations”: [ {

“branch”: <block_hash>, “data”: <hex encoded>

} … ], “check_signature”: <boolean> // optional

}

Returns

Json object

class pytezos.rpc.helpers.PreapplyBlockQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(block, sort=None, timestamp=None)[source]

Simulate the validation of a block that would contain the given operations and return the resulting fitness and context hash.

Parameters
  • block

    Json input {

    ”protocol_data”: {

    “protocol”: “Pt24m4xiPbLDhVgVfABUjirbmda3yohdN82Sp9FeuAXJ4eV9otd”, “priority”: <integer>, “proof_of_work_nonce”: <hex encoded> “seed_nonce_hash”: <Base58 encoded>, // optional “signature”: <Base58 encoded>

    }, “operations”: [ [ {

    ”protocol”: “Pt24m4xiPbLDhVgVfABUjirbmda3yohdN82Sp9FeuAXJ4eV9otd”, “branch”: <block_hash>, “contents”: [ {} … ], // kind-specific “signature”: <Base58 encoded>

    }… ] … ]

    }

  • sort – by what?

  • timestamp – timestamp

Returns

Json object

class pytezos.rpc.helpers.PreapplyOperationsQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(operations)[source]

Simulate the validation of operation(s).

Parameters

operations

Json input [{

”protocol”: “Pt24m4xiPbLDhVgVfABUjirbmda3yohdN82Sp9FeuAXJ4eV9otd”, “branch”: <block_hash>, “contents”: [ {} … ], // kind-specific “signature”: <Base58 encoded>

}]

Returns

Json array (operations with metadata)

class pytezos.rpc.helpers.ScriptsEntrypoint(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(script, entrypoint=None)[source]

Return the type of the given entrypoint.

Parameters
  • script – Micheline expression

  • entrypoint – Name of the entrypoint (leave None if there’s only one)

class pytezos.rpc.helpers.ScriptsEntrypoints(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(script)[source]

Return the list of entrypoints of the given script.

Parameters

script – Micheline expression

class pytezos.rpc.helpers.ScriptsPackDataQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(expression)[source]

Computes the serialized version of some data expression using the same algorithm as script instruction PACK.

Parameters

expression

Json input {

”data”: <michelson expression>, “type”: <michelson expression>, “gas”: <bignum> // optional

}

Returns

Packed data (hex encoded)

class pytezos.rpc.helpers.ScriptsRunCodeQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(invocation)[source]

Run a piece of code in the current context.

Parameters

invocation

Json input {

”script”: <michelson expression>, “storage”: <michelson expression>, “entrypoint”: <string>, “input”: <michelson expression>, “chain_id”: <base58 encoded>, “amount”: <mutez>, “source”: <account address>, // optional “payer”: <account address>, // optional “gas”: <bignum> // optional

}

Returns

Resulting storage, spawned operations, and bigmap diff

class pytezos.rpc.helpers.ScriptsRunOperationQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(operation)[source]

Run an operation without signature checks.

Parameters

operation

Json input {

”branch”: <block_hash>, “contents”: [ {} … ], // kind-specific “signature”: <Base58 encoded>

}

Returns

Json object (operation with metadata)

class pytezos.rpc.helpers.ScriptsTraceCodeQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(invocation)[source]

Run a piece of code in the current context, keeping a trace.

Parameters

invocation

Json input {

”script”: <michelson expression>, “storage”: <michelson expression>, “entrypoint”: <string>, “input”: <michelson expression>, “chain_id”: <base58 encoded>, “amount”: <mutez>, “source”: <account address>, // optional “payer”: <account address>, // optional “gas”: <bignum> // optional

}

Returns

Resulting storage, spawned operations, bigmap diff, and trace

class pytezos.rpc.helpers.ScriptsTypecheckCodeQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(expression)[source]

Check that some data expression is well formed and of a given type in the current context.

Parameters

expression

Json input {

”program”: <michelson expression>, “gas”: <bignum> // optional

}

Returns

Json object

class pytezos.rpc.helpers.ScriptsTypecheckDataQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
post(expression)[source]

Check that some data expression is well formed and of a given type in the current context.

Parameters

expression

Json input {

”data”: <michelson expression>, “type”: <michelson expression>, “gas”: <bignum> // optional

}

Returns

Json object

Interval search

class pytezos.rpc.search.BlockSliceQuery(start: int, stop=None, **kwargs)[source]
__call__() → list[source]

Get block hashes (base58) for this interval.

find_ballots() → Generator[source]

Find ballot operations for the current period.

Returns

Generator (lazy)

find_operation(operation_group_hash) → dict[source]

Find operation by hash.

Parameters

operation_group_hash – base58

Raises

StopIteration if not found

find_origination(contract_id)[source]

Find contract origination.

Parameters

contract_id – Contract ID (KT-address)

find_proposal_injection(proposal_id)[source]

Find proposal injection.

Parameters

proposal_id – Proposal hash (base58)

find_upvotes(proposal_id) → Generator[source]

Find upvoting operations for the given proposal.

Parameters

proposal_id – Proposal hash (base58)

Returns

Generator (lazy)

get_range()[source]

Get block level range.

property path
class pytezos.rpc.search.CyclesQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)[source]

Get current cycle.

property path
class pytezos.rpc.search.PeriodQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)

Call self as a function.

property path
class pytezos.rpc.search.VotingPeriodsQuery(node: pytezos.rpc.node.RpcNode, path='', caching=False, params=None, timeout=None)[source]
__call__(**params)[source]

Get current voting period.

property path
pytezos.rpc.search.find_state_change(head: int, last: int, get: Callable, equals: Callable, pred_value: Any) -> (<class 'int'>, typing.Any)[source]
pytezos.rpc.search.find_state_change_intervals(head: int, last: int, get: Callable, equals: Callable, step=60) → Generator[source]
pytezos.rpc.search.find_state_changes(head: int, last: int, get: Callable, equals: Callable, step=60) → Generator[source]
pytezos.rpc.search.walk_state_change_interval(head: int, last: int, get: Callable, equals: Callable, head_value: Any, last_value: Any) → Generator[source]

Node wrapper and errors

exception pytezos.rpc.node.RpcError[source]
args
classmethod from_errors(errors: list)[source]
classmethod from_response(res: requests.models.Response)[source]
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

class pytezos.rpc.node.RpcMultiNode(uri, network)[source]
delete(path, params=None)
get(path, params=None, caching=False, cache_key=None, timeout=None)
post(path, params=None, json=None, caching=False)
put(path, params=None)
request(method, path, **kwargs) → requests.models.Response[source]
class pytezos.rpc.node.RpcNode(uri, network='')[source]
delete(path, params=None)[source]
get(path, params=None, caching=False, cache_key=None, timeout=None)[source]
post(path, params=None, json=None, caching=False)[source]
put(path, params=None)[source]
request(method, path, **kwargs) → requests.models.Response[source]
pytezos.rpc.node.gen_error_variants(error_id) → list[source]
pytezos.rpc.node.urljoin(*args)[source]
exception pytezos.rpc.errors.MichelsonBadContractParameter[source]

Either no parameter was supplied to a contract with a non-unit parameter type, a non-unit parameter was passed to an account, or a parameter was supplied of the wrong type

exception pytezos.rpc.errors.MichelsonBadReturn[source]

Unexpected stack at the end of a lambda or script

exception pytezos.rpc.errors.MichelsonRuntimeError[source]

Catch all michelson_v1 errors

exception pytezos.rpc.errors.MichelsonScriptRejected[source]

A FAILWITH instruction was reached

exception pytezos.rpc.errors.TezArithmeticError[source]

Catch all tez errors