RPC query engine
Shell-level endpoints
- class pytezos.rpc.shell.BlockInjectionQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function BlockInjectionQuery.post> 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'). block format: .. code-block:: python { "data": <hex-encoded block header>, "operations": [ [ { "branch": <block_hash>, "data": <hex-encoded operation> } ... ] ... ] } :param block: JSON input :param _async: By default, the RPC will wait for the block to be validated before answering,set True if you don't want to. :param force: :param chain: Optionally you can specify the chain (main/test) :returns: ID of the block [source]
- class pytezos.rpc.shell.ChainQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- class pytezos.rpc.shell.ConnectionQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- class pytezos.rpc.shell.DescribeQuery(node: RpcNode, path: str = '', 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: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- class pytezos.rpc.shell.MempoolQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- class pytezos.rpc.shell.MonitorQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
-
- property path
- class pytezos.rpc.shell.NetworkItems(node: RpcNode, path: str = '', params=None, timeout=None)[source]
-
- property path
- class pytezos.rpc.shell.NetworkLogQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
-
- property path
- class pytezos.rpc.shell.OperationInjectionQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function OperationInjectionQuery.post> 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. :param operation: Hex-encoded operation data or bytes :param _async: By default, the RPC will wait for the operation to be (pre-)validated before answering, set True if you don't want to. :param chain: Optionally you can specify the chain :returns: ID of the operation [source]
- class pytezos.rpc.shell.PendingOperationsQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- class pytezos.rpc.shell.ProtocolInjectionQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function ProtocolInjectionQuery.post> Inject a protocol in node. protocol format: .. code-block:: python { "expected_env_version": <integer>, "components": [{ "name": <unistring>, "interface"?: <hex-encoded data>, "implementation": <hex-encoded data> } ... ]} } :param protocol: JSON input :param _async: :param force: :returns: ID of the protocol [source]
- class pytezos.rpc.shell.ShellQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property block: BlockQuery
Cached head block, useful if you just want to explore things.
- property blocks: BlocksQuery
Shortcut for chains.main.blocks
- property contracts
Shortcut for head.context.contracts
- property cycles
Operate on cycles rather than blocks.
- get_confirmations = <function ShellQuery.get_confirmations> Returns the number of blocks applied after the operation was included in chain :param opg_hash: Operation group hash :param kind: Operation kind ('transaction', 'origination', etc) :param branch: Block ID one should stop the search at :param head: Block ID one should start the search from :return: Number of confirmations (0 if not found) [source]
- property head: BlockQuery
Shortcut for blocks.head
- property mempool
Shortcut for chains.main.mempool
- property path
- property voting_periods
Operate on voting periods rather than blocks.
- wait_blocks = <function ShellQuery.wait_blocks> Iterates over future blocks (waits and yields block hash), handles reorgs :param current_block_hash: hash of the current block (head) :param max_blocks: number of blocks to iterate (not including the current one) :param yield_current: yield current block hash at the very beginning :param time_between_blocks: override protocol constant :param block_timeout: set block timeout (by default Pytezos will wait for a long time) :return: block hashes [source]
- wait_next_block = <function ShellQuery.wait_next_block> Wait until next block is finalized. :param prev_hash: Current block hash (optional). If not set, current head is used. :param time_between_blocks: override the corresponding parameter from constants :param max_iterations: Manually set the number of iterations :param delay_sec: Sleep delay .. deprecated:: 3.2.2 This will be removed in 4.0.0. Use wait_blocks() instead[source]
- wait_operations = <function ShellQuery.wait_operations> Wait for one or many operations gain enough confirmations :param opg_hashes: list of operation hashes :param ttl: max time-to-live value (in mempool) :param min_confirmations: minimum number of blocks after inclusion to wait for :param current_block_hash: current block hash (head) :param time_between_blocks: override protocol constant :param block_timeout: set block timeout (by default Pytezos will wait for a long time) :return: list of operation contents with metadata [source]
Protocol-specific endpoints
- class pytezos.rpc.protocol.BigMapGetQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function BigMapGetQuery.post> Access the value associated with a key in the big map storage of the michelson. query format: .. code-block:: python { 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) :param query: JSON input :returns: Micheline expression [source]
- class pytezos.rpc.protocol.BlockQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property baker
Query block producer (baker).
- Return type:
- property path
- property predecessor
Query previous block.
- Return type:
- class pytezos.rpc.protocol.BlocksQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(length=1, head=None, min_date=None)[source]
List 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:
- property current_voting_period
Get block range for the current voting period.
- Return type:
- 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: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- class pytezos.rpc.protocol.ContractQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- count = <function ContractQuery.count> Get contract counter iterator: it returns incremented value on each call.[source]
- property path
- class pytezos.rpc.protocol.EndorsingPower(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function EndorsingPower.post> Get the endorsing power of an endorsement operation, that is, the number of slots that the op has. endorsement_operation format: .. code-block:: python { "branch": $block_hash, "contents": [ $operation.alpha.contents ... ], "signature": $Signature } :param endorsement_operation [source]
- class pytezos.rpc.protocol.OperationListListQuery(node: RpcNode, path: str = '', 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 = <function OperationListListQuery.find_ballots> Find operations of kind `ballot`. :param proposal_id: Proposal hash (optional) :returns: list of operation contents [source]
- find_origination = <function OperationListListQuery.find_origination> Find origination of the contract. :param contract_id: Contract ID (KT-address) :returns: operation content [source]
- find_upvotes = <function OperationListListQuery.find_upvotes> Find operations of kind `proposal` for given proposal. :param proposal_id: Proposal hash (base58) :returns: list of operation contents [source]
- 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: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- class pytezos.rpc.protocol.ProposalQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
-
- property path
RPC Helpers
- class pytezos.rpc.helpers.BakingRightsQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(level=None, cycle=None, delegate=None, max_priority=None, _all=None)[source]
Retrieve 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: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function ForgeBlockHeaderQuery.post> Forge block header. block_header format: .. code-block:: python { "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> } :param block_header: JSON input :returns: { "block": <hex encoded> } [source]
- class pytezos.rpc.helpers.ForgeOperationsQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- class pytezos.rpc.helpers.ForgeProtocolDataQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function ForgeProtocolDataQuery.post> Forge the protocol-specific part of a block header. protocol_data format: .. code-block:: python { "priority": <integer>, "nonce_hash": <Base58 nonce hash>, // optional "proof_of_work_nonce": <hex encoded> // optional } :param protocol_data: JSON input :returns: { "protocol_data": <hex encoded> } [source]
- class pytezos.rpc.helpers.ParseBlockQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function ParseBlockQuery.post> Retrieve protocol-specific part of a block header and signature. block_header format: .. code-block:: python { "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> } :param block_header: JSON input :returns: JSON object [source]
- class pytezos.rpc.helpers.ParseOperationsQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function ParseOperationsQuery.post> Extract contents and signatures from the forged and optionally signed operations (bulk). operations format: .. code-block:: python { "operations": [ { "branch": <block_hash>, "data": <hex encoded> } ... ], "check_signature": <boolean> // optional } :param operations: JSON input :returns: JSON object [source]
- class pytezos.rpc.helpers.PreapplyBlockQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function PreapplyBlockQuery.post> Simulate the validation of a block that would contain the given operations and return the resulting fitness and context hash. block format: .. code-block:: python { "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> }... ] ... ] } :param block: JSON input :param sort: by what? :param timestamp: timestamp :returns: JSON object [source]
- class pytezos.rpc.helpers.PreapplyOperationsQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function PreapplyOperationsQuery.post> Simulate the validation of operation(s). operations format: .. code-block:: python [{ "protocol": "Pt24m4xiPbLDhVgVfABUjirbmda3yohdN82Sp9FeuAXJ4eV9otd", "branch": <block_hash>, "contents": [ {} ... ], // kind-specific "signature": <Base58 encoded> }] :param operations: JSON input :returns: JSON array (operations with metadata) [source]
- class pytezos.rpc.helpers.ScriptsEntrypoint(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- class pytezos.rpc.helpers.ScriptsEntrypoints(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- class pytezos.rpc.helpers.ScriptsPackDataQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function ScriptsPackDataQuery.post> Compute the serialized version of some data expression using the same algorithm as script instruction PACK. expression format: .. code-block:: python { "data": <michelson expression>, "type": <michelson expression>, "gas": <bignum> // optional } :param expression: JSON input :returns: Packed data (hex encoded) [source]
- class pytezos.rpc.helpers.ScriptsRunCodeQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function ScriptsRunCodeQuery.post> Run a piece of code in the current context. invocation format: .. code-block:: python { "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 } :param invocation: JSON input :returns: Resulting storage, spawned operations, and bigmap diff [source]
- class pytezos.rpc.helpers.ScriptsRunOperationQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function ScriptsRunOperationQuery.post> Run an operation without signature checks. operation format: .. code-block:: python { "branch": <block_hash>, "contents": [ {} ... ], // kind-specific "signature": <Base58 encoded> } :param operation: JSON input :returns: JSON object (operation with metadata) [source]
- class pytezos.rpc.helpers.ScriptsRunScriptViewQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function ScriptsRunScriptViewQuery.post> Simulate a call to a michelson view invocation format: .. code-block:: python { "contract": <address>, "view": <string>, "input": <michelson expression>, "unlimited_gas": <boolean>, "chain_id": <base58 encoded>, "source": <account address>, // optional "payer": <account address>, // optional "gas": <bignum>, // optional "unparsing_mode": <Readable | Optimized | Optimized_legacy>, "now": <integer>, // optional "level": <integer>, // optional } :param invocation: JSON input :returns: Resulting storage, spawned operations, and bigmap diff [source]
- class pytezos.rpc.helpers.ScriptsTraceCodeQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function ScriptsTraceCodeQuery.post> Run a piece of code in the current context, keeping a trace. invocation format: .. code-block:: python { "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 } :param invocation: JSON input :returns: Resulting storage, spawned operations, bigmap diff, and trace [source]
- class pytezos.rpc.helpers.ScriptsTypecheckCodeQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function ScriptsTypecheckCodeQuery.post> Check that some data expression is well formed and of a given type in the current context. expression format: .. code-block:: python { "program": <michelson expression>, "gas": <bignum> // optional } :param expression: JSON input :returns: JSON object [source]
- class pytezos.rpc.helpers.ScriptsTypecheckDataQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- post = <function ScriptsTypecheckDataQuery.post> Check that some data expression is well formed and of a given type in the current context. expression format: .. code-block:: python { "data": <michelson expression>, "type": <michelson expression>, "gas": <bignum> // optional } :param expression: JSON input :returns: JSON object [source]
Interval search
- class pytezos.rpc.search.BlockSliceQuery(start: int, stop=None, **kwargs)[source]
-
- find_ballots = <function BlockSliceQuery.find_ballots> Find ballot operations for the current period. :returns: Generator (lazy) [source]
- find_operation = <function BlockSliceQuery.find_operation> Find operation by hash. :param operation_group_hash: base58 :raises: StopIteration if not found [source]
- find_origination = <function BlockSliceQuery.find_origination> Find contract origination. :param contract_id: Contract ID (KT-address) [source]
- find_proposal_injection = <function BlockSliceQuery.find_proposal_injection> Find proposal injection. :param proposal_id: Proposal hash (base58) [source]
- find_upvotes = <function BlockSliceQuery.find_upvotes> Find upvoting operations for the given proposal. :param proposal_id: Proposal hash (base58) :returns: Generator (lazy) [source]
- property path
- class pytezos.rpc.search.CyclesQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
-
- property path
- class pytezos.rpc.search.PeriodQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
- __call__(**params)
Call self as a function.
- property path
- class pytezos.rpc.search.VotingPeriodsQuery(node: RpcNode, path: str = '', params=None, timeout=None)[source]
-
- property path
- pytezos.rpc.search.find_state_change(head: int, last: int, get: ~typing.Callable, equals: ~typing.Callable, pred_value: ~typing.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]
Node wrapper and errors
- exception pytezos.rpc.node.RpcError[source]
- args
- classmethod from_errors(errors: List[Dict[str, Any]]) RpcError [source]
Create RpcError from “errors” section of response JSON.
- with_traceback()
Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.
- class pytezos.rpc.node.RpcMultiNode(uri: str | List[str])[source]
Request proxy for multiple nodes chosen for each request in round-robin order.
- delete(path: str, params: Dict[str, Any] | None = None, timeout: int | None = None) Response
- get(path: str, params: Dict[str, Any] | None = None, timeout: int | None = None) Response
- post(path: str, params: Dict[str, Any] | None = None, json=None, timeout: int | None = None) Response | str
- put(path: str, params: Dict[str, Any] | None = None, timeout: int | None = None) Response
- class pytezos.rpc.node.RpcNode(uri: str | List[str], headers: Dict[str, str] | None = None)[source]
Request proxy for a single Tezos node.
- delete(path: str, params: Dict[str, Any] | None = None, timeout: int | None = None) Response [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