src.escrow.blockchain package


src.escrow.blockchain.golos_blockchain module

Module contents

class src.escrow.blockchain.BaseBlockchain[source]

Bases: abc.ABC

Abstract class to represent blockchain node client for escrow exchange.

address = None

Address used by bot.

assets = frozenset()

Frozen set of assets supported by blockchain.

check_timeout(offer_id: bson.objectid.ObjectId) → None[source]

Start transaction check timeout asynchronously.

Parameters:offer_id_id of escrow offer.
check_transaction(*, offer_id: bson.objectid.ObjectId, from_address: str, amount_with_fee: decimal.Decimal, amount_without_fee: decimal.Decimal, asset: str, memo: str, transaction_time: float) → bool[source]

Check transaction in history of escrow address.

  • offer_id_id of escrow offer.
  • from_address – Address which sent assets.
  • amount_with_fee – Amount of transferred asset with fee added.
  • amount_without_fee – Amount of transferred asset with fee substracted.
  • asset – Transferred asset.
  • memo – Memo in blockchain transaction.
  • transaction_time – Start of transaction check.

Queue member with timeout handler or None if queue member is timeouted.


Close connection with blockchain node.

connect() → None[source]

Establish connection with blockchain node.

create_queue() → List[Dict[str, Any]][source]

Create queue from unconfirmed transactions in database.

explorer = '{}'

Template of URL to transaction in blockchain explorer. Should contain {} which gets replaced with transaction id.

get_limits(asset: str) → src.escrow.blockchain.InsuranceLimits[source]

Get maximum amounts of asset which will be insured during escrow exchange.

Escrow offer starts only if sum of it doesn’t exceed these limits.

get_min_time(queue: List[Dict[str, Any]]) → float[source]

Get timestamp of earliest transaction from queue.

is_block_confirmed(block_num: int, op: Mapping[str, Any]) → bool[source]

Check if block # block_num has op after confirmation.

Check block on blockchain-specific conditions to consider it confirmed.

  • block_num – Number of block to check.
  • op – Operation to check.
name = None

Internal name of blockchain referenced in config.ESCROW_FILENAME.


Get list of node URLs.

schedule_timeout(queue_member: Dict[str, Any]) → Optional[Dict[str, Any]][source]

Schedule timeout of transaction check.

transfer(to: str, amount: decimal.Decimal, asset: str, memo: str = '') → str[source]

Transfer asset from self.address.

  • to – Address assets are transferred to.
  • amount – Amount of transferred asset.
  • asset – Transferred asset.

URL to transaction in blockchain explorer.

trx_url(trx_id: str) → str[source]

Get URL on transaction with ID trx_id on explorer.


Get private key encoded to WIF.

exception src.escrow.blockchain.BlockchainConnectionError[source]

Bases: Exception

Unsuccessful attempt at connection to blockchain node.

class src.escrow.blockchain.InsuranceLimits[source]

Bases: tuple

Maximum amount of insured asset.


Limit on sum of a single offer.


Limit on overall sum of offers.

class src.escrow.blockchain.StreamBlockchain[source]

Bases: src.escrow.blockchain.BaseBlockchain

Blockchain node client supporting continuous stream to check transaction.


Add transaction to self._queue to be checked.

Same parameters as in self.check_transaction.

check_timeout(offer_id: bson.objectid.ObjectId) → None[source]

Start transaction check timeout asynchronously.

Parameters:offer_id_id of escrow offer.
remove_from_queue(offer_id: bson.objectid.ObjectId) → Optional[Mapping[str, Any]][source]

Remove transaction with specified offer_id value from self._queue.

Parameters:offer_id_id of escrow offer.
Returns:True if transaction was found and False otherwise.
start_streaming() → None[source]

Start streaming in background asynchronous task.

stream() → None[source]

Stream new blocks and check if they contain transactions from self._queue.

Use built-in method to subscribe to new blocks if node has it, otherwise get new blocks in blockchain-specific time interval between blocks.

If block contains desired transaction, call self._confirmation_callback. If it returns True, remove transaction from self._queue and stop streaming if self._queue is empty.

exception src.escrow.blockchain.TransferError[source]

Bases: Exception

Unsuccessful attempt at transfer.