tonpy.blockscanner package

Submodules

tonpy.blockscanner.blockscanner module

class tonpy.blockscanner.blockscanner.BlockScanner(lcparams: dict, start_from: int = None, load_to: int = None, nproc: int = 20, loglevel: int = 0, raw_process: Callable = None, chunk_size: int = 100, tx_chunk_size: int = 5000, out_queue: Queue = None, only_mc_blocks: bool = False, parse_txs_over_ls: bool = False, transaction_subscriptions: CustomSubscription = None, account_subscriptions: CustomAccountSubscription = None, database_provider: BaseDatabaseProvider = None, emulate_before_output: bool = None, live_load_enable: bool = False, load_chunks: List[Tuple[int, int]] = None, allow_skip_mc_in_live: bool = True, blocks_to_load: List[BlockIdExt] = None, dton_key: str = None)

Bases: Thread

Concurrent block scanner for TON blockchain.

Loads masterchain and shardchain blocks, optionally filters/emulates transactions, and emits results to a queue or a user-provided callback.

detect_cs_p(list_)
load_historical()

Load historical data from start_from to load_to using chunked processing.

load_live()
load_mcs(from_, to_)
load_process_blocks(blocks, tx_subscriptions=None)
load_process_shard(known_shards, stop_shards)
prepare_key_blocks(shards_data, mc_data)
prepare_prev_block_data(shards_data, mc_data)
run()

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

tonpy.blockscanner.blockscanner.convert_account_blocks_to_txs(account_blocks_all)

Convert raw account_blocks VmDict to a mapping of account->list of transactions.

Args:

account_blocks_all: Cell with account blocks dictionary.

Returns:

dict: Mapping account address (int) to list of tx info dicts with keys: tx, lt, now, account, is_tock.

tonpy.blockscanner.blockscanner.get_mega_libs(dton_key, num_try=100)

Fetch the mega_libs list from dton.io public API with retries.

Args:

dton_key: Reserved for future API auth (not used at the moment). num_try: Number of retry attempts.

Returns:

dict | list: API response data if successful.

tonpy.blockscanner.blockscanner.load_process_blocks(blocks_chunk='__no__default__', lcparams='__no__default__', loglevel='__no__default__', emulate_before_output='__no__default__', tx_subscriptions='__no__default__')
tonpy.blockscanner.blockscanner.load_process_shard_chunk(shards_chunk='__no__default__', known_shards='__no__default__', stop_shards='__no__default__', lcparams='__no__default__', loglevel='__no__default__', parse_txs_over_ls=False, load_only_known=False)
tonpy.blockscanner.blockscanner.process_block(block, lc, emulate_before_output, tx_subscriptions)

Load and filter transactions for a given block, optionally emulating state.

Args:

block: Dict with block metadata and optionally pre-extracted account_blocks. lc: LiteClient instance. emulate_before_output: If True, load previous account states for emulation. tx_subscriptions: Optional subscription filter for transactions.

Returns:

list: List of (block_dict, account_state_cell_or_none, txs_for_account) triples.

tonpy.blockscanner.blockscanner.process_mc_blocks(seqnos='__no__default__', lcparams='__no__default__', loglevel='__no__default__', parse_txs_over_ls='__no__default__')
tonpy.blockscanner.blockscanner.process_shard(x, prev_data=None, lc=None, loglevel=None, known_shards=None, stop_shards=None, parse_txs_over_ls=False, load_only_known=False)
tonpy.blockscanner.blockscanner.process_subscriptions(data='__no__default__', tx_subscriptions: CustomSubscription = None, account_subscriptions: CustomAccountSubscription = None)

Filter or process transactions against provided subscriptions.

Args:

data: Tuple of (block_dict, account_state_cell or None, txs_per_account). tx_subscriptions: Subscription to filter transactions by predicate. account_subscriptions: Subscription for account-based processing (not yet implemented).

Returns:

List of [block_id, tx_cell] pairs when account_state is None.

tonpy.blockscanner.blockscanner.raw_process(chunk)

tonpy.blockscanner.database_provider module

class tonpy.blockscanner.database_provider.BaseDatabaseProvider

Bases: object

Please note that BaseDatabaseProvider must be design like DB connector, because it’ll be pickle and moved to other process in Pool

class tonpy.blockscanner.database_provider.DummyDatabaseProvider

Bases: BaseDatabaseProvider

tonpy.blockscanner.subscriptions module

class tonpy.blockscanner.subscriptions.AccountSubscription(code_hash: str | int | None = None)

Bases: CustomSubscription

check(account: CellSlice) → bool
class tonpy.blockscanner.subscriptions.AndCustomSubscription(left: CustomSubscription, right: CustomSubscription)

Bases: object

check(*args, **kwargs) → bool
class tonpy.blockscanner.subscriptions.CustomAccountSubscription

Bases: object

check(workchain: int, address: str, account_state: CellSlice | None = None) → bool
Parameters:

  • workchain

  • address

  • account_state – Will provided only if emulate_before_output is True

Returns:

class tonpy.blockscanner.subscriptions.CustomSubscription

Bases: object

check(tx: CellSlice) → bool
class tonpy.blockscanner.subscriptions.OrCustomSubscription(left: CustomSubscription, right: CustomSubscription)

Bases: object

check(*args, **kwargs) → bool
class tonpy.blockscanner.subscriptions.TransactionSubscription(op_code: int | None = None, text: str | None = None)

Bases: CustomSubscription

check(tx: CellSlice) → bool

tonpy.blockscanner.txs_inblock_emulator module

tonpy.blockscanner.txs_inblock_emulator.emulate_in_block(data)

Module contents